package com.test.service;

import com.test.Utils.StockTradeMapper;
import com.test.entity.StockTrade;
import com.test.entity.TradeResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import redis.clients.jedis.Jedis;
import org.apache.flink.util.Collector;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Slf4j
public class StockTradeAnalysis {
    public static int number = 0 ;
    public static int number_ = 0 ;
    public static FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
            .setHost("192.168.43.150")
            .setPort(6379)
            .build();

    // 自定义Sink函数
    public static class AccumulatingRedisSink extends RichSinkFunction<TradeResult> {
        private transient Jedis jedis;
        private final FlinkJedisPoolConfig config;
        private String type;

        public AccumulatingRedisSink(FlinkJedisPoolConfig config, String type) {
            this.config = config;
            this.type = type;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            jedis = new Jedis(config.getHost(), config.getPort());
        }

        @Override
        public void close() throws Exception {
            if (jedis != null) {
                jedis.close();
            }
            super.close();
        }

        @Override
        public void invoke(TradeResult value, Context context) throws Exception {
            double totalAmount;
            double totalVolumeCount;
            if (type.equals("1s")) {
                String existingValue = jedis.lindex("1:", -1);
                if (existingValue != null) {
                    String[] parts = existingValue.split(":");
                    totalAmount = Double.parseDouble(parts[0]);
                    totalVolumeCount = Double.parseDouble(parts[1]);
                } else {
                    totalAmount = 0.0;
                    totalVolumeCount = 0L;
                }

                totalAmount += Double.parseDouble(value.getCount());
                totalVolumeCount += Double.parseDouble(value.getTradeVolume());
                if(StockTradeAnalysis.number_<20){
                    System.out.println("这是1s"+totalAmount);
                    System.out.println("这是1s"+totalVolumeCount);
                    number_++;
                }
                jedis.rpush("1:", totalAmount + ":" + totalVolumeCount);
            }
        }

    }

    // 定义全局聚合函数
    public static class GlobalTradeAggregator implements AggregateFunction<StockTrade, Map<String, TradeAccumulator>, Map<String, TradeAccumulator>> {
        @Override
        public Map<String, TradeAccumulator> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public Map<String, TradeAccumulator> add(StockTrade value, Map<String, TradeAccumulator> accumulator) {


            String stockCode = value.getStockCode();
            TradeAccumulator acc = accumulator.getOrDefault(stockCode, new TradeAccumulator(stockCode));

            acc.totalAmount += value.getTradeVolume();
            acc.totalVolumeCount++;


            accumulator.put(stockCode, acc);

            return accumulator;
        }

        @Override
        public Map<String, TradeAccumulator> getResult(Map<String, TradeAccumulator> accumulator) {
            TradeAccumulator.type++;

            return accumulator;
        }

        @Override
        public Map<String, TradeAccumulator> merge(Map<String, TradeAccumulator> a, Map<String, TradeAccumulator> b) {
            for (Map.Entry<String, TradeAccumulator> entry : b.entrySet()) {
                String stockCode = entry.getKey();
                TradeAccumulator accB = entry.getValue();
                TradeAccumulator accA = a.getOrDefault(stockCode, new TradeAccumulator(stockCode));
                accA.totalAmount += accB.totalAmount;
                accA.totalVolumeCount += accB.totalVolumeCount;
                a.put(stockCode, accA);
            }
            return a;
        }
    }

    // 定义全局结果函数
public static class GlobalTradeResultFunction implements AllWindowFunction<Map<String, TradeAccumulator>, TradeResult, TimeWindow> {
    private final String windowType;

    public GlobalTradeResultFunction(String windowType) {
        this.windowType = windowType;
    }

    @Override
    public void apply(TimeWindow window, Iterable<Map<String, TradeAccumulator>> input, Collector<TradeResult> out) throws Exception {
        // 创建累加器来存储所有股票代码的总金额和总成交量
        double totalAmount = 0.0;
        long totalVolumeCount = 0L;

        for (Map<String, TradeAccumulator> accumulatorMap : input) {
            for (Map.Entry<String, TradeAccumulator> entry : accumulatorMap.entrySet()) {
                TradeAccumulator accumulator = entry.getValue();
                totalAmount += accumulator.totalAmount;
                totalVolumeCount += accumulator.totalVolumeCount;
            }
        }

        // 收集累加后的结果
        if (StockTradeAnalysis.number < 20) {
            System.out.println(
                "Processing window for all stocks, " +
                "window type: " + windowType +
                ", window: [" + window.getStart() + " - " + window.getEnd() +
                "], total amount: " + totalAmount +
                ", total volume count: " + totalVolumeCount +
                "，这是第" + TradeAccumulator.type + "次"
            );
            StockTradeAnalysis.number++;
        }

        out.collect(new TradeResult("all_stocks", String.valueOf(totalAmount), String.valueOf(totalVolumeCount)));
    }
}


    // 定义累加器
    public static class TradeAccumulator {
        public static int type = 0;
        public String stockCode;
        public double totalAmount = 0.0;
        public long totalVolumeCount = 0L;

        public TradeAccumulator(String stockCode) {
            this.stockCode = stockCode;
        }
    }

    public static void createDataStream(StreamExecutionEnvironment env) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.88.135:19092,192.168.88.135:29092,192.168.88.135:39092");
        properties.setProperty("group.id", "stock-trade-analysis");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
            "stock-trades-",
            new SimpleStringSchema(),
            properties
        );
        consumer.setStartFromEarliest();
        consumer.setCommitOffsetsOnCheckpoints(true);

        // 创建数据流并添加水印
        DataStream<StockTrade> tradeStream = env
                .addSource(consumer)
                .map(new StockTradeMapper())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<StockTrade>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getTime())
                );

        // 一秒
        tradeStream
                .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(1000)))
                .aggregate(new GlobalTradeAggregator(), new GlobalTradeResultFunction("1s"))
                .addSink(new AccumulatingRedisSink(redisConfig,"1s"));
//        tradeStream
//                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
//                .trigger(EventTimeTrigger.create())
//                .aggregate(new GlobalTradeAggregator(), new GlobalTradeResultFunction("1h"))
//                .addSink(new AccumulatingRedisSink(redisConfig,"1h"));

        log.info("Stock trade analysis configured");
    }
}






